#include "globalarray.h"

MPI_Comm UO_COMM_ASK;
MPI_Comm UO_COMM_RESP;
MPI_Comm UO_COMM_FINALIZE;

MPI_Datatype msgtype;

int self= -1;
int rank_size = -1;

int default_load_balance = MODE_RANDOM_BALANCE;
int memLoads[MAX_NODES], loadRank[MAX_NODES];
int lastNode;

pthread_cond_t init_cond;
pthread_mutex_t mpi_mut;
pthread_mutex_t init_mut;

Msg recv_msg;
Msg send_msg;

int signal_finalize = 0;
int first_time = 1;


void defineMsgStruct()
{
	MPI_Datatype oldtypes[6];
	MPI_Aint offsets[6],extent;
	int blockcounts[6];
	
	// .type, .dim, .length .loc .offset
	offsets[0] = 0;
	oldtypes[0] = MPI_INT;
	blockcounts[0] = 5;
	
	// .value
	MPI_Type_extent(MPI_INT, &extent);
	offsets[1] = 5 * extent;
	oldtypes[1] = MPI_DOUBLE;
	blockcounts[1] = 1;
	
	// .index[]
	MPI_Type_extent(MPI_DOUBLE, &extent);
	offsets[2] = 1 * extent+ offsets[1];
	oldtypes[2] = MPI_INT;
	blockcounts[2] = 3;
	
	// .size[]
	MPI_Type_extent(MPI_INT, &extent);
	offsets[3] = 3 * extent+ offsets[2];
	oldtypes[3] = MPI_INT;
	blockcounts[3] = 3;
	
	// .name[]
	MPI_Type_extent(MPI_INT, &extent);
	offsets[4] = 3 * extent+ offsets[3];
	oldtypes[4] = MPI_CHAR;
	blockcounts[4] = 16;
	
	// .value_array[10]+void*
	MPI_Type_extent(MPI_CHAR, &extent);
	offsets[5] =  16* extent+ offsets[4];
	oldtypes[5] = MPI_DOUBLE;
	blockcounts[5] = 21;
	
	MPI_Type_struct(6, blockcounts, offsets, oldtypes, &msgtype);
	MPI_Type_commit(&msgtype);
	
	return;
}

void uo_init(int argc, char **argv)
{
    int flag,i;
    int split_color = 1;
    int split_rank;
    MPI_Status status;
	
	int provided;
    MPI_Initialized(&flag);
    if (!flag) {
        MPI_Init(&argc, &argv);
        //MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided);
        //printf ("asked: %d, got: %d\n", MPI_THREAD_MULTIPLE, provided);
    }
	
    int comm_world_self;
    MPI_Comm_rank(MPI_COMM_WORLD, &comm_world_self);
    split_rank = comm_world_self;
    
	
    MPI_Comm_dup(MPI_COMM_WORLD, &UO_COMM_ASK);
    MPI_Comm_dup(MPI_COMM_WORLD, &UO_COMM_RESP);
    MPI_Comm_dup(MPI_COMM_WORLD, &UO_COMM_FINALIZE);
    MPI_Barrier(UO_COMM_FINALIZE);
    
    MPI_Comm_size(UO_COMM_RESP, &rank_size);
    MPI_Comm_rank(UO_COMM_RESP, &self);
    
    srand(self);
	
    lastNode = -1;
    for(i=0; i<rank_size; i++){
        memLoads[i] = 0;
        loadRank[i] = i;
    }
    
    defineMsgStruct();
    
    printf("self: %d, size: %d\n", self, rank_size);
    
    pthread_mutex_init(&mpi_mut, NULL);
    pthread_mutex_init(&init_mut, NULL);
    pthread_cond_init(&init_cond, NULL);
    
    pthread_mutex_lock(&init_mut);
    
    pthread_t listenthread;
    pthread_create(&listenthread, NULL, listener, NULL);
    
    pthread_cond_wait(&init_cond, &init_mut);
    pthread_mutex_unlock(&init_mut);
    
    return;
}

void *listener()
{
	int finalize_count = 0, i;
    pthread_mutex_lock(&init_mut);
	
    printf("[Listener %d] Initializing listener...\n", self);
    
	
    while(1) {
        MPI_Status status;
        MPI_Request request;
        printf("[Listener %d] Looping: I am still alive! %d\n",self, finalize_count);
        
        if (first_time == 1) {
            printf("[Listener %d] Waiting for initializing barrier...\n", self);
            MPI_Barrier(UO_COMM_ASK);
            pthread_cond_signal(&init_cond);
            pthread_mutex_unlock(&init_mut);
            first_time = 0;
        }
		
        if (self == 0)
        {
            uo_recv(&recv_msg, 1, msgtype, MPI_ANY_SOURCE, MPI_ANY_TAG, UO_COMM_ASK, &status);
			//#ifdef _DEBUG 
            printf("[Listener %d, TAG %d] receive a msg: type:%d dim:%d len:%d loc:%d value:%f index[0]:%d index[1]:%d index[2]:%d size[0]:%d size[1]:%d size[2]:%d name:%s\n",
				   self,status.MPI_TAG,recv_msg.type,recv_msg.dim,recv_msg.length,recv_msg.loc,recv_msg.value,
				   recv_msg.index[0],recv_msg.index[1],recv_msg.index[2],
				   recv_msg.size[0],recv_msg.size[1],recv_msg.size[2],recv_msg.name);
			//#endif
            
            int command = tag_to_cmd(status.MPI_TAG);
			if (command == CMD_NEW_ARRAY) {
			    int tag = status.MPI_TAG;
			    int index = hashPut(recv_msg.name, recv_msg.type, recv_msg.size, recv_msg.dim);
			    printf("index=%d\n",index);
			    if(index==-1){
					printf("String %s already exists",recv_msg.name);
			    }else elements[index].loc = uo_load_balance(index);
				
			    send_msg.type = recv_msg.type;
			    send_msg.dim = recv_msg.dim;
			    send_msg.loc = elements[index].loc;
				printf("[Listener %d, TAG %d] CMD_NEW_ARRAY array loc:%d\n",self,status.MPI_TAG,elements[index].loc); 
				for(i=0; i<recv_msg.dim; i++)
					send_msg.size[i] = recv_msg.size[i];
			    strcpy(send_msg.name, recv_msg.name);
				
                printf("[Listener %d, TAG %d] Sending back to %d\n",self,status.MPI_TAG,status.MPI_SOURCE);
				
				uo_send(&send_msg, 1, msgtype, status.MPI_SOURCE, status.MPI_TAG, UO_COMM_RESP);
			    printf("[Listener %d, TAG %d] Finish sending back to %d\n",self,status.MPI_TAG,status.MPI_SOURCE);
			}
            else if (command == CMD_PUT_ELEMENT) {
				//#ifdef _DEBUG
                printf("[Listener %d, TAG %d] receive put_element msg:  value:%f index[0]:%d index[1]:%d index[2]:%d name:%s\n",
					   self,status.MPI_TAG,recv_msg.value,
					   recv_msg.index[0],recv_msg.index[1],recv_msg.index[2],
					   recv_msg.name);
                //#endif
				
                int hash_index = hashGet(recv_msg.name);
                recv_msg.loc = elements[hash_index].loc;
                
                printf("PUT: master recv request dim:%d name:%s[%d][%d][%d] loc:%d\n",
					   elements[hash_index].dim, recv_msg.name,
					   recv_msg.index[0],recv_msg.index[1],recv_msg.index[2],recv_msg.loc); 
				
                //MPI_Send(&recv_msg, 1, msgtype, status.MPI_SOURCE, status.MPI_TAG, UO_COMM_RESP);
                uo_send(&recv_msg, 1, msgtype,  status.MPI_SOURCE,status.MPI_TAG , UO_COMM_RESP);
                printf("PUT:master send loc of name:%s[%d][%d][%d]  to %d\n",
					   recv_msg.name,recv_msg.index[0],recv_msg.index[1],recv_msg.index[2],recv_msg.loc);    
                printf("[Listener %d, TAG %d] respond location msg to node %d\n",
					   self, status.MPI_TAG, status.MPI_SOURCE);
				
            } else if (command == CMD_PUT_ARRAY) {
				int tag = status.MPI_TAG;
				int index = hashGet(recv_msg.name);
				send_msg.loc = elements[index].loc;
				uo_send(&send_msg, 1, msgtype, status.MPI_SOURCE, tag, UO_COMM_RESP);
            } else if (command == CMD_GET_ELEMENT) {
				printf("[Listener %d, TAG %d] receive request from %d\n", self, status.MPI_TAG, status.MPI_SOURCE);
                int hash_index = hashGet(recv_msg.name);
                recv_msg.loc = elements[hash_index].loc;
                recv_msg.dim = elements[hash_index].dim;
				
                printf("GET: master send to source:%d  tag:%d  UO_COMM_RESP\n",status.MPI_SOURCE, status.MPI_TAG);
                //MPI_Send(&recv_msg, 1, msgtype, status.MPI_SOURCE, status.MPI_TAG, UO_COMM_RESP);
                uo_send(&recv_msg, 1, msgtype, status.MPI_SOURCE,status.MPI_TAG , UO_COMM_RESP);
                printf("[Listener %d, TAG %d] respond location msg to node %d\n",
					   self, status.MPI_TAG, status.MPI_SOURCE);
				
			} else if (command == CMD_GET_ARRAY) {
				int index = hashGet(recv_msg.name);
				send_msg.loc=elements[index].loc;
				uo_send(&send_msg, 1, msgtype, status.MPI_SOURCE, status.MPI_TAG, UO_COMM_RESP);
            } else if (command == CMD_FREE_ARRAY) {
				int index = hashGet(recv_msg.name);
				send_msg.loc=elements[index].loc;
				uo_send(&send_msg, 1, msgtype, status.MPI_SOURCE, status.MPI_TAG, UO_COMM_RESP);
				elements[index].name[0]='\0';
            } else if (command == CMD_FINALIZE) {
                ++finalize_count;
                printf("[Listener %d] Finalize count = %d...\n", self, finalize_count);
                if (finalize_count == rank_size) break;
            } else {
				printf("[Listener %d, TAG %d] Unknown command!\n", self, status.MPI_TAG);
            }
        }
        else  // self is not 0
        {
            uo_recv(&recv_msg, 1, msgtype, MPI_ANY_SOURCE, MPI_ANY_TAG, UO_COMM_ASK, &status);
			
            int command = tag_to_cmd(status.MPI_TAG);
            if(command == CMD_NEW_ARRAY){
 	            int sizet = 1,bytes,i;
			    for(i=0; i<recv_msg.dim; i++){
				    sizet *= recv_msg.size[i];
			    }
				
			    switch(recv_msg.type){
			        case TYPE_INT: bytes = sizeof(int); break;
			        case TYPE_CHAR: bytes = sizeof(char); break;
			        case TYPE_FLOAT: bytes = sizeof(float); break;
			        case TYPE_DOUBLE: bytes = sizeof(double); break;
			        default:;
			    }
			    void* addr = malloc(sizet*bytes);
				printf("CREATE ARRAY size:%d ,[%d %d %d]\n",sizet,recv_msg.size[0],recv_msg.size[1],recv_msg.size[2]);
				
			    int index = hashPut(recv_msg.name, recv_msg.type, recv_msg.size, recv_msg.dim);
			    elements[index].addr = addr;
			}else if (command == CMD_PUT_ELEMENT) {
				// storing an element
                printf("[Listener %d, TAG %d] store data for node %d array:%s value:%f\n",
					   self, status.MPI_TAG,recv_msg.loc, recv_msg.name,recv_msg.value);
				
                if (recv_msg.loc == self) {
                    // store it
                    storeElement(recv_msg);
                    
                } else {
					printf("[Listener %d, TAG %d] destination loc is wrong\n", self, status.MPI_TAG);
				}
			}else if (command == CMD_PUT_ARRAY) {
				int hash_index = hashGet(recv_msg.name);
				
				void* addr = elements[hash_index].addr;
				
				for(i=0 ; i<recv_msg.length; i++){
					switch(recv_msg.type){
						case TYPE_INT: *((int*)addr+i+recv_msg.offset) = (int)recv_msg.value_array[i];break;
						case TYPE_CHAR: *((char*)addr+i+recv_msg.offset) = (char)recv_msg.value_array[i];break;
						case TYPE_FLOAT: *((float*)addr+i+recv_msg.offset) = (float)recv_msg.value_array[i];break;
						case TYPE_DOUBLE: *((double*)addr+i+recv_msg.offset) = (double)recv_msg.value_array[i];break;
							
					}
					//printf("%d ",((int*)addr)[i+recv_msg.offset]);	
				}
				
				//printf("\n");
			} else if (command == CMD_GET_ELEMENT) {
				
				if (recv_msg.loc == self) {
					
                    int tag = status.MPI_TAG;
                    int to = status.MPI_SOURCE; 
					
                    // sent value tp api: get_element
                    sentElement(recv_msg,to,tag);
					
                    printf("[Listener %d TAG %d] send value to node %d\n",
						   self, status.MPI_TAG, status.MPI_SOURCE );
                }
			} else if (command == CMD_GET_ARRAY) {
				int hash_index = hashGet(recv_msg.name);
				void* addr = elements[hash_index].addr;
				for(i=0 ; i<recv_msg.length; i++){
					switch(recv_msg.type){
						case TYPE_INT: recv_msg.value_array[i] = (double)(*((int*)addr+i+recv_msg.offset));break;
						case TYPE_CHAR: recv_msg.value_array[i] = (double)(*((char*)addr+i+recv_msg.offset));break;
						case TYPE_FLOAT: recv_msg.value_array[i] = (double)(*((float*)addr+i+recv_msg.offset));break;
						case TYPE_DOUBLE: recv_msg.value_array[i] = (double)(*((double*)addr+i+recv_msg.offset));break;
					}
					
				}
				uo_send(&recv_msg, 1, msgtype, status.MPI_SOURCE, status.MPI_TAG, UO_COMM_RESP);
            }else if(command == CMD_FREE_ARRAY){
			    int hash_index = hashGet(recv_msg.name);
				void* addr = elements[hash_index].addr;
				free(addr);
				elements[hash_index].name[0]='\0';
			}else if (command == CMD_FINALIZE) {
                ++finalize_count;
                printf("[Listener %d] Finalize count = %d...\n", self, finalize_count);
                if (finalize_count == rank_size) break;
            } else {
                printf("[Listener %d, TAG %d] Unknown command!\n", self, status.MPI_TAG);
            } 
        }
    } // while
    
    //finalizing steps
    printf("[Listener %d] Waiting for barrier...\n", self);
	
    pthread_mutex_lock(&mpi_mut);
    MPI_Barrier(UO_COMM_FINALIZE);
    pthread_mutex_unlock(&mpi_mut);
    printf("[Listener %d] Barrier overcome...\n", self);
    
    signal_finalize = 1;
    printf("[Listener %d] Exiting Listener\n", self);
    
    return;
}

void storeElement(Msg recv_msg)
{
	int hash_index = hashGet(recv_msg.name);
	int type = elements[hash_index].type;
	
	printf("PUT: store to getHash name:%s[%d][%d][%d] dim=%d  addr:%p\n",elements[hash_index].name,
		   recv_msg.index[0],recv_msg.index[1],recv_msg.index[2],
		   elements[hash_index].dim, elements[hash_index].addr);
	
	printf("PUT: name:%s size %d %d %d\n",recv_msg.name,elements[hash_index].size[0],
		   elements[hash_index].size[1],
		   elements[hash_index].size[2]); 
	int i,j,k;
	i = recv_msg.index[0];
	j = recv_msg.index[1];
	k = recv_msg.index[2];
	int sizei,sizej,sizek;
	sizei = elements[hash_index].size[0];
	sizej = elements[hash_index].size[1];
	sizek = elements[hash_index].size[2];
	printf("sizei:%d sizej:%d sizek:%d\n", sizei,sizej,sizek);
	
	int offset;
	if (elements[hash_index].dim == 1 )
	{
		offset = i;
	}
	if (elements[hash_index].dim == 2 ) 
	{
		offset  = i*sizej+j; 
	}
	if (elements[hash_index].dim == 3 )
	{
		offset  = i*sizej*sizek+j*sizek +k;
	}
	
	switch (type)
	{
			int c,d;
		case TYPE_INT:
			((int *)elements[hash_index].addr)[offset]  = (int)recv_msg.value;
			printf("PUT: store to name:%s dim:%d offset:%d  a[%d][%d][%d]=%d\n",recv_msg.name,elements[hash_index].dim,
				   offset,i,j,k,((int *)elements[hash_index].addr)[offset]);
			
			
			break;
		case TYPE_FLOAT:
			((float *)elements[hash_index].addr)[offset]  = (float)recv_msg.value;
			printf("PUT: store to name:%s  dim:%d   a[%d][%d][%d]=%f\n",recv_msg.name,elements[hash_index].dim,
				   i,j,k,((float *)elements[hash_index].addr)[offset]);
			
			
			break;
		case TYPE_DOUBLE:
			((double *)elements[hash_index].addr)[offset]  = recv_msg.value;
			printf("PUT:store to name:%s dim:%d  offset:%d [%d][%d][%d]=%f\n",
				   recv_msg.name,elements[hash_index].dim,
				   offset,i,j,k,((double *)elements[hash_index].addr)[offset]);
			
#ifdef _DEBUG
			for (c = 0; c <sizei; c ++)
			{
				for (d = 0; d <sizej;d++)
				{
					printf("a[%d][%d]:%f\t",c,d,
						   ((double *)elements[hash_index].addr)[c*sizei+d]);
				}
				printf("\n");
			}
#endif
			
			break;
		default:
			printf("PUT_ELEMENT: type not handled\n");
			break;
	}   
	
	return;
}

void sentElement(Msg recv_msg,int to, int tag)
{
	int hash_index = hashGet(recv_msg.name);
	int type = elements[hash_index].type;
	
	printf("SENT getHash array %s dim=%d  addr:%p recv_msg.value(before sent):%.0lf\n",
		   elements[hash_index].name,
		   elements[hash_index].dim, elements[hash_index].addr,recv_msg.value);
	
	int i,j,k;
	i = recv_msg.index[0];
	j = recv_msg.index[1];
	k = recv_msg.index[2];
	int sizei,sizej,sizek;
	sizei = elements[hash_index].size[0];
	sizej = elements[hash_index].size[1];
	sizek = elements[hash_index].size[2];
	printf("GET sizei:%d sizej:%d sizek:%d\n", sizei,sizej,sizek);
	
	int offset;
	if (elements[hash_index].dim == 1 )
	{
		offset = i;
	}
	if (elements[hash_index].dim == 2 )
	{
		offset  = i*sizej+j;
	}
	if (elements[hash_index].dim == 3 )
	{
		offset  = i*sizej*sizek+j*sizek +k;
	}
	
	switch (type)
	{
			int c,d;
		case TYPE_INT:
			recv_msg.value = ((int *)elements[hash_index].addr)[offset] ;
			printf("SENT dim:%d offset:%d  a[%d][%d][%d]=%d\n",elements[hash_index].dim,
				   offset,i,j,k,((int *)elements[hash_index].addr)[offset]);
			
			break;
		case TYPE_FLOAT:
			recv_msg.value = ((float *)elements[hash_index].addr)[offset];
			printf("SENT dim:%d   a[%d][%d][%d]=%f\n",elements[hash_index].dim,
				   i,j,k,((float *)elements[hash_index].addr)[offset]);
			
			break;
		case TYPE_DOUBLE:
			recv_msg.value = ((double *)elements[hash_index].addr)[offset] ;
			printf("SENT dim:%d  offset:%d a[%d][%d][%d]=%f recv_msg.value:%f\n",
				   elements[hash_index].dim, offset,i,j,k,
				   ((double *)elements[hash_index].addr)[offset],recv_msg.value);
			
#ifdef _DEBUG
			for (c = 0; c <sizei; c ++)
			{
				for (d = 0; d <sizej;d++)
				{
					printf("a[%d][%d]:%f\t",c,d,
						   ((double *)elements[hash_index].addr)[c*sizei+d]);
				}
				printf("\n");
			}
#endif
			
			break;
		default:
			printf("PUT_ELEMENT: type not handled\n");
			break;
	}
	printf("GET:worker send dim:%d name:%s[%d][%d][%d] value:%f to %d tag:%d UO_COMM_RESP\n",
		   elements[hash_index].dim,recv_msg.name,
		   recv_msg.index[0],recv_msg.index[1],recv_msg.index[2],
		   recv_msg.value,to,tag);
	//MPI_Send(&recv_msg, 1, msgtype, to, tag, UO_COMM_RESP);
	uo_send(&recv_msg, 1, msgtype, to, tag, UO_COMM_RESP);
	
	
	return; 
}

void uo_new_array(char* name, int type, int *size, int dimension)
{
    printf("[API %d] called uo_new_array().\n",self);
	int put_tag, i;
	MPI_Status status;
	Msg send0_msg,recv0_msg;
	Msg sendx_msg;
	send0_msg.type = type;
	send0_msg.dim = dimension;
	for(i=0; i<dimension; i++)
		send0_msg.size[i]=size[i];
	strcpy(send0_msg.name, name);
	put_tag = cmd_to_tag(CMD_NEW_ARRAY);
	printf("[Node %d] Send MSG_NEW_ARRAY to 0\n",self);
	
	
	uo_send(&send0_msg, 1, msgtype, 0, put_tag, UO_COMM_ASK);
	
	printf("[Node %d] Waiting for 0 for TAG %d\n", self, put_tag);
	
	uo_recv(&recv0_msg, 1, msgtype, 0, put_tag, UO_COMM_RESP, &status);
	
	printf("[Node %d] Receive from 0\n", self);
	sendx_msg.type = type;
	sendx_msg.dim = dimension;
	sendx_msg.loc = recv0_msg.loc;
	for(i=0;i<dimension;i++)
		sendx_msg.size[i]=size[i];
	strcpy(sendx_msg.name,name);
	printf("[Node %d] Send MSG_NEW_ARRAY to %d\n",self,sendx_msg.loc);
	put_tag = cmd_to_tag(CMD_NEW_ARRAY);
	uo_send(&sendx_msg, 1, msgtype, sendx_msg.loc, put_tag, UO_COMM_ASK);
	
	return;
}

void uo_put_element(char *name, int *index, void* value, int type)
{
	MPI_Status status;
	/* ask master where to put element */
	int to;
	int put_tag;
	int prev_put_tag;
	int *int_value;
	float *float_value;
	double *double_value;
	
	
	Msg put_msg;
	Msg put_msg_loc; // response msg from master with location
	put_msg.type = type;
	put_msg.dim = -1;
	put_msg.length = -1;
	put_msg.loc = -1; // wait for master node to update
	
	
	switch (type)
	{
		case TYPE_INT:
			put_msg.value = *(int*)value;
			break;
			
		case TYPE_FLOAT:
			put_msg.value = *(float*)value;
			break;
			
		case TYPE_DOUBLE:
			//printf("value:%f\n", *(double*)value);
			put_msg.value = *(double*)value;
			break;
		default:
			printf("put_element: type unknown\n");
			break;
    }
	
	put_msg.index[0] = *index;
	put_msg.index[1] = *(index+1);
	put_msg.index[2] = *(index+2);
	
	strcpy(put_msg.name, name);
	put_tag = cmd_to_tag(CMD_PUT_ELEMENT);
	
	//MPI_Send(&put_msg,1,msgtype,0,put_tag,UO_COMM_ASK);
	uo_send(&put_msg, 1, msgtype, 0,put_tag,UO_COMM_ASK);
	
	printf("[API %d TAG %d] send put_msg to master\n",self,put_tag);
	printf("PUT: send put_msg to master: type:%d   value:%f name:%s[%d][%d][%d]\n",
		   put_msg.type,put_msg.value,put_msg.name,put_msg.index[0],put_msg.index[1],put_msg.index[2] );
	
	
	//MPI_Recv(&put_msg_loc, 1, msgtype, 0, put_tag, UO_COMM_RESP, &status);
	uo_recv(&put_msg_loc, 1, msgtype, 0, put_tag, UO_COMM_RESP, &status);
	
	printf("[API %d TAG %d] recv put from master tag:%d UO_COMM_RESP\n",self,put_tag,put_tag);
	
	printf("PUT: recv loc from master: type:%d  value:%f name:%s[%d][%d][%d]\n",
		   put_msg_loc.type,put_msg_loc.value,
		   put_msg_loc.name,put_msg_loc.index[0],put_msg_loc.index[1],put_msg_loc.index[2]);
	
	printf("[API %d TAG %d] uo_put_element receive response from master\n", self, put_tag);
	prev_put_tag = put_tag;
	put_tag = cmd_to_tag(CMD_PUT_ELEMENT);
	to = put_msg_loc.loc;
	//MPI_Send(&put_msg_loc,1,msgtype,to,put_tag,UO_COMM_ASK);
	uo_send(&put_msg_loc, 1, msgtype, to,put_tag,UO_COMM_ASK);
	
	printf("PUT: send to loc, type:%d  value:%f name:%s[%d][%d][%d]\n",
		   put_msg_loc.type,put_msg_loc.value,
		   put_msg_loc.name,put_msg_loc.index[0],put_msg_loc.index[1],put_msg_loc.index[2]);
	
	printf("[API %d TAG %d  prev_TAG %d] uo_put_element storing value %f at node %d\n",
		   self, put_tag, prev_put_tag,put_msg_loc.value,to);
	
	
	return;
}

void uo_put_array(char *name, int start, void *buffer, int buffer_start, int size, int type)
{
	printf("[API %d] called uo_put_array().\n", self);
	// Send msg to the listener at node 0  
	int i,j;
	MPI_Status status;
	Msg send0_msg, recv0_msg;
	int put_tag;
	
	send0_msg.dim = 1;
	send0_msg.type = type;
	strcpy(send0_msg.name, name); 
	put_tag = cmd_to_tag(CMD_PUT_ARRAY);
    
	uo_send(&send0_msg, 1, msgtype, 0, put_tag, UO_COMM_ASK);
	uo_recv(&recv0_msg, 1, msgtype, 0, put_tag, UO_COMM_RESP, &status);
    
	send0_msg.loc = recv0_msg.loc; 
	send0_msg.type = type;
	
	i=j=0;
	int k=0;
	
	while(1){
		//printf("%d ",((int*)buffer)[buffer_start+i]);
		
		switch(send0_msg.type){
			case TYPE_INT: send0_msg.value_array[j] = (double)((int*) buffer)[buffer_start+i]; break;
			case TYPE_CHAR: send0_msg.value_array[j] = (double)((char*) buffer)[buffer_start+i];break;
			case TYPE_FLOAT: send0_msg.value_array[j] = (double)((float*) buffer)[buffer_start+i];break;    
			case TYPE_DOUBLE: send0_msg.value_array[j] = (double)((double*) buffer)[buffer_start+i];break;
		}
		i++;
		j++;
		if(i==size || j==20){
			send0_msg.length = j;
			send0_msg.offset = start+i-j;
			uo_send(&send0_msg, 1, msgtype, send0_msg.loc , put_tag, UO_COMM_ASK);
			if(i==size)break;
			j=0;
		}
	} 
	
	return;
}

void uo_get_element(char *name, int *index, void *value,int type)
{
	MPI_Status status;
	int to;
	int get_tag;
	
	int* int_value;
	float* float_value;
	double* double_value;
	
	Msg get_msg;
	Msg get_msg_loc; // response msg from master with location
	Msg recv_msg; // response from location node
	get_msg.type = type;
	get_msg.loc = -1; // wait for master node to update
	get_msg.value = -1;
	get_msg.index[0] = *index;
	get_msg.index[1] = *(index+1);
	get_msg.index[2] = *(index+2);
	strcpy(get_msg.name, (char*)name);
	get_tag = cmd_to_tag(CMD_GET_ELEMENT);
	
	
	//MPI_Send(&get_msg,1,msgtype,0,get_tag,UO_COMM_ASK);
	uo_send(&get_msg, 1, msgtype, 0, get_tag, UO_COMM_ASK);
	
	printf("GET: api ask master tag:%d UO_COMM_ASK\n",get_tag);   
	
	printf("[API %d TAG %d] send get_msg to master\n",self,get_tag);
#ifdef _DEBUG
	printf("[API %d TAG %d] send get_msg to master: type:%d dim:%d len:%d loc:%d value:%f index[0]:%d index[1]:%d index[2]:%d  size[0]:%d size[1]:%d size[2]:%d name:%s\n",
		   self,get_tag,
		   get_msg.type,get_msg.dim,get_msg.length,get_msg.loc,get_msg.value,
		   get_msg.index[0],get_msg.index[1],get_msg.index[2],
		   get_msg.size[0],get_msg.size[1],get_msg.size[2],name );
#endif
	
	printf("GET:api expect master tag:%d UO_COMM_RESP\n",get_tag);
	//MPI_Recv(&get_msg_loc, 1, msgtype, 0, get_tag, UO_COMM_RESP, &status);
	uo_recv(&get_msg_loc, 1, msgtype, 0, get_tag, UO_COMM_RESP, &status);
	printf("GET:api recv master tag:%d UO_COMM_RESP\n",get_tag);
	
	printf("[API %d TAG %d] get_element receive response from master\n", self, get_tag);
	
	get_tag = cmd_to_tag(CMD_GET_ELEMENT);
	to = get_msg_loc.loc;
	
	//MPI_Send(&get_msg_loc,1,msgtype,to,get_tag,UO_COMM_ASK);
	uo_send(&get_msg_loc, 1, msgtype, to, get_tag, UO_COMM_ASK);
	printf("GET:api ask for value dim:%d a[%d][%d][%d] from node:%d tag:%d UO_COMM_ASK\n",
		   get_msg_loc.dim,get_msg_loc.index[0],get_msg_loc.index[1],get_msg_loc.index[2],to,get_tag);
	printf("[API %d TAG %d] get_msg  to node %d\n",self, get_tag, to);
	
	printf("GET:api expect value  dim:%d a[%d][%d][%d] from node:%d tag:%d UO_COMM_RESP\n",
		   get_msg_loc.dim,get_msg_loc.index[0],get_msg_loc.index[1],get_msg_loc.index[2],to,get_tag); 
	//MPI_Recv(&recv_msg, 1, msgtype, to, get_tag, UO_COMM_RESP, &status);
	uo_recv(&recv_msg, 1, msgtype, to, get_tag, UO_COMM_RESP, &status);
	
	
	printf("GET: api  recv_msg.value :%f tag:%d  name:%s[%d][%d][%d]  UO_COMM_RESP\n",recv_msg.value,get_tag,
           recv_msg.name,recv_msg.index[0],recv_msg.index[1],recv_msg.index[2]);
	switch (type)
    {
		case TYPE_INT:
			*((int *)value) = (int)recv_msg.value;
			printf("GOT: return value :%d\n",*((int *)value));
			break;
		case TYPE_FLOAT:
			*((float *)value) = (float)recv_msg.value;
			printf("GOT: return value :%f\n",*((float *)value));
			break;
		case TYPE_DOUBLE:
			*((double *)value) = recv_msg.value;
			printf("GOT: return value :%f\n",*((double *)value));
			break;  
		default:
			printf("GET_ELEMENT: type:%d not handled\n",type); 
			break;
			
	}
	
	printf("[API %d TAG %d] get_msg getting data %f from %d \n",self, get_tag, recv_msg.value, to);
	return;
}

void uo_get_array(char *name, int start, void *buffer, int buffer_start, int size, int type)
{
	Msg send0_msg, recv0_msg;
    MPI_Status status;
	strcpy(send0_msg.name, name);
	send0_msg.offset = start;
	
	int get_tag = cmd_to_tag(CMD_GET_ARRAY);
	uo_send(&send0_msg, 1, msgtype, 0, get_tag, UO_COMM_ASK);
	uo_recv(&recv0_msg, 1, msgtype, 0, get_tag, UO_COMM_RESP, &status);
	send0_msg.loc = recv0_msg.loc;
	
	int i,j;
	i=j=0;
	while(1){
		i++;
		j++;
		if(i==size || j==20){
			send0_msg.length = j;
			send0_msg.offset = start+i-j;
			//printf("length=%d,offset=%d\n",j,start+i-j);
			uo_send(&send0_msg, 1, msgtype, send0_msg.loc, get_tag, UO_COMM_ASK);
			uo_recv(&recv0_msg, 1, msgtype, send0_msg.loc, get_tag, UO_COMM_RESP, &status);
			int k;
			//printf("Message length=%d\n",send0_msg.length);
			for(k=0 ; k<send0_msg.length; k++)
				switch (type) {
					case TYPE_INT:
						((int*)buffer)[send0_msg.offset+k] = (int)recv0_msg.value_array[k];
						break;
					case TYPE_CHAR:
						((char*)buffer)[send0_msg.offset+k] = (char)recv0_msg.value_array[k];
						break;
					case TYPE_FLOAT:
						((float*)buffer)[send0_msg.offset+k] = (float)recv0_msg.value_array[k];
						break;
					case TYPE_DOUBLE:
						((double*)buffer)[send0_msg.offset+k] = (double)recv0_msg.value_array[k];
					default:
						break;
				}
			if(i==size)break;
			j=0;
		}
	}
	return;
}

void uo_free_array(char *name)
{
	printf("[API %d] Used uo_free_array\n",self);
	int tag = cmd_to_tag(CMD_FREE_ARRAY);
	Msg send0_msg, recv0_msg;
	MPI_Status status;
	strcpy(send0_msg.name, name);
	
	uo_send(&send0_msg, 1, msgtype, 0, tag, UO_COMM_ASK);
    uo_recv(&recv0_msg, 1, msgtype, 0, tag, UO_COMM_RESP, &status);
	
	send0_msg.loc=recv0_msg.loc;
	uo_send(&send0_msg, 1, msgtype, send0_msg.loc, tag, UO_COMM_ASK);
}

void uo_finalize()
{
    printf("[API %d] Enter finalizing...\n", self);
	
	//    Msg finalize_msg;
    Msg finalize_msg[rank_size];
    int finalize_tag = cmd_to_tag(CMD_FINALIZE);
    
    int fin_i = (self + 1) % rank_size;;
    int fin_count;
    for (fin_count = 0; fin_count < rank_size; fin_count++) {
		uo_send(&finalize_msg[fin_i], 1, msgtype, fin_i, finalize_tag, UO_COMM_ASK);
        printf("[API %d TAG %d] Sent finalize signal to %d\n", self, finalize_tag, fin_i);
        fin_i = (fin_i + 1) % rank_size;
    }
    
    printf("[API %d] Waiting for finalize signal...\n", self);
    
    while (signal_finalize == 0) {
        sched_yield();
    }
    
    MPI_Finalize();
    printf("[API %d] Exiting UO GA\n", self);
    return;
}

void uo_recv(Msg* msg, int size, MPI_Datatype type, int source, int tag, MPI_Comm comm, MPI_Status* status)
{
	int request_flag = 0;
	MPI_Request request;
	pthread_mutex_lock(&mpi_mut);
	MPI_Irecv(msg, size, type, source, tag, comm, &request);
	pthread_mutex_unlock(&mpi_mut);
	do {
		pthread_mutex_lock(&mpi_mut);
		MPI_Test(&request, &request_flag, status);
		pthread_mutex_unlock(&mpi_mut);
		if (!request_flag) {
			sched_yield();
		} else {
			break;
		}
	} while(1);
	
}

void uo_send(Msg* msg, int size, MPI_Datatype type, int source, int tag, MPI_Comm comm)
{
	pthread_mutex_lock(&mpi_mut);	
	MPI_Send(msg, size, type, source, tag, comm);
	pthread_mutex_unlock(&mpi_mut);
}


int cmd_to_tag(int cmd)
{
    return cmd + (rand() % 1000);
}

int tag_to_cmd(int tag)
{
    return (tag / 1000)*1000;
}

int uo_load_balance(int index)
{
	switch(default_load_balance){
		case MODE_ROTATE_BALANCE:
			if(lastNode==-1)lastNode=0;
			else lastNode++;
			if(lastNode==rank_size)lastNode=1;
			return lastNode; 
		case MODE_RANDOM_BALANCE:
			return rand()%(rank_size-1)+1;
		case MODE_LEAST_BALANCE:{ 
			int sizet=1,bytes,i;
			for(i=0;i<elements[index].dim;i++){
				sizet*=elements[index].size[i];
			}
			switch(elements[index].type){
				case TYPE_INT:bytes = sizet * sizeof(int);break;
				case TYPE_CHAR:bytes = sizet * sizeof(char);break;
				case TYPE_FLOAT:bytes = sizet * sizeof(float);break;
				case TYPE_DOUBLE:bytes = sizet * sizeof(double);break;
				default:;
			}
			memLoads[loadRank[1]]+=bytes;
			i=1;
			while(i+1!=rank_size && memLoads[loadRank[i]]>memLoads[loadRank[i+1]]){
				int tmp = loadRank[i];
				loadRank[i] = loadRank[i+1];
				loadRank[i+1] = tmp;
			}
			return i;} 
		case MODE_DEFAULT_BALANCE:
			return 1;
	}
}

